package com.hudsonmq.spring.impl.produce.impl;


import com.hudson.codec.domain.Message;
import com.hudson.codec.domain.RemotingCommand;
import com.hudson.common.domain.Duration;
import com.hudson.common.enums.MessageTypeEnum;
import com.hudson.core.concurrent.CallBack;
import com.hudsonmq.spring.HudsonMQClientController;
import com.hudsonmq.spring.domain.produce.TopicQueueInfo;
import com.hudsonmq.spring.impl.produce.RemoteClient;
import com.hudsonmq.spring.impl.produce.helper.SendMessageHelper;
import com.hudsonmq.spring.impl.produce.helper.SendMessageSyncHelper;
import com.hudsonmq.spring.utils.NettyUtil;
import com.hudsonmq.spring.utils.SnowflakeIdGeneratorUtil;
import io.netty.channel.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.StandardCharsets;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

public class SyncRemoteClient implements RemoteClient {


    private static final Logger log = LoggerFactory.getLogger(SyncRemoteClient.class);

    private static final int MAX_RETRY_NUM = 3;

    @Override
    public Boolean send(Message message, CallBack callBack) {
        long messageId = SnowflakeIdGeneratorUtil.getMessageId();

        TopicQueueInfo topicQueue = SendMessageHelper.sendMessage(messageId, message, callBack);

        CompletableFuture<Boolean> nettyFuture = new CompletableFuture<>();
        SendMessageSyncHelper.put(messageId, nettyFuture);


        try {
            return nettyFuture.get(3, TimeUnit.SECONDS);
        } catch (TimeoutException e) {
            int retryNum = 0;
            log.error("获取同步信息超时,重试{}", messageId);
            if (topicQueue != null) {
                while (retryNum < MAX_RETRY_NUM) {
                    if (retrySend(messageId, topicQueue, message, callBack, retryNum)) {
                        log.info("消息重试成功");
                        return true;
                    }
                    retryNum++;
                }
            }
        } catch (Exception e) {
            log.error("消息 {} 发送失败", messageId);
        } finally {
            SendMessageSyncHelper.removeFuture(messageId);
        }
        return false;
    }

    @Override
    public Boolean relaySend(Message message, CallBack callBack, Duration duration) {
        long messageId = SnowflakeIdGeneratorUtil.getMessageId();

        TopicQueueInfo topicQueue = SendMessageHelper.sendRelayMessage(messageId, message, duration);

        CompletableFuture<Boolean> nettyFuture = new CompletableFuture<>();
        SendMessageSyncHelper.put(messageId, nettyFuture);

        nettyFuture.whenComplete((result, ex) -> {
            SendMessageSyncHelper.removeFuture(messageId);
        });

        try {
            return nettyFuture.get(3, TimeUnit.SECONDS);
        } catch (TimeoutException e) {
            int retryNum = 0;
            log.error("获取同步信息超时,重试{}", messageId);
            if (topicQueue != null) {
                while (retryNum < MAX_RETRY_NUM) {
                    if (retrySend(messageId, topicQueue, message, callBack, retryNum)) {
                        return true;
                    }
                    retryNum++;
                }
            }
        } catch (Exception e) {
            throw new RuntimeException("发送失败进行重试");
        }
        return null;
    }

    @Override
    public boolean retrySend(long messageId, TopicQueueInfo topicQueue, Message message, CallBack callBack, int retryNum) {
        Channel chanel = HudsonMQClientController.BROKER_CHANNEL_MAP.get(topicQueue.getBrokerName());
        message.setQueueId(topicQueue.getQueueId());

        NettyUtil.write(chanel, RemotingCommand.builder()
                .type(MessageTypeEnum.SEND_MESSAGE_REQ.getType())
                .opaque(messageId)
                .body(Message.serialize(message))
                .build()).addListener(f -> {
            if (f.isSuccess()) {
                log.debug("消息 {} 發送成功", messageId);
                if (callBack != null) {
                    SendMessageSyncHelper.put(messageId, callBack);
                }
            } else {
                log.error("消息{}:{}发送失败", messageId, new String(message.getBody(), StandardCharsets.UTF_8));
            }
        });

        CompletableFuture<Boolean> nettyFuture = new CompletableFuture<>();
        SendMessageSyncHelper.put(messageId, nettyFuture);

        try {
            return nettyFuture.get(3, TimeUnit.SECONDS);
        } catch (TimeoutException e) {
            log.error("message:{},第{}次重试失败", messageId, retryNum);
        } catch (Exception e) {
            log.error("message:{},第{}次发送失败", messageId, retryNum);
        } finally {
            SendMessageSyncHelper.removeFuture(messageId);
        }
        return false;
    }

}
